/*  Layer 4 RingBuffer
 *  Copyright (C) 2016 ZHANG X. <201560039@uibe.edu.cn>
 *  __ |  \   /   __| 
 *    /     /    |    
 *  ___|  _/ _\ \___| 
 *
 *  **All Rights Reserved**
 */
#include "l4rb.h"

#define L4RB_LISTEN_EV INACTIVE_CONN

int
l4rb_queue_init(struct L4rb *l4rb, uint32_t max_conn){
    l4rb->max_conn = UINT32_MAX;
    /*  Round down to power of two  */
    while(l4rb->max_conn > max_conn) l4rb->max_conn >>= 1;
    l4rb->queue = epoll_create(l4rb->max_conn + 1);
    for(register uint32_t curr=0; curr< l4rb->max_conn; curr++)
        l4rb->conn[curr].fd = FD_DOWN;
    return l4rb->queue;
}
inline int
l4rb_bind_ipv4(int socket, struct in_addr addr, in_port_t port){
    struct sockaddr_in ipv4addr = {
        .sin_family = AF_INET,
        .sin_addr = addr,
        .sin_port = port,
    };
    return bind(socket, (struct sockaddr *)&ipv4addr, sizeof(ipv4addr));
}
inline int
l4rb_bind_ipv6(int socket, struct in6_addr addr, in_port_t port){
    struct sockaddr_in6 ipv6addr = {
        .sin6_family = AF_INET6,
        .sin6_addr = addr,
        .sin6_port = port,
    };
    return bind(socket, (struct sockaddr *)&ipv6addr, sizeof(ipv6addr));
}
int
l4rb_listen(struct L4rb *l4rb, char *conn_str){
    unsigned char addr[sizeof(struct in6_addr)];
    in_port_t port;
    int server_socket;
    char *port_str = NULL;
    if((port_str = strchr(conn_str, ']'))){
        conn_str++;
        port_str[0] = '\0';
        port_str += 2;
    }else if(strchr(conn_str, '.') && (port_str = strchr(conn_str, ':'))){
        port_str[0] = '\0';
        port_str += 1;
    }
    if(port_str)
        port = htons((uint16_t)atoi(port_str));
    else port = 0;

    if(inet_pton(AF_INET, conn_str, addr) > 0){
        /*  IPv4    */
        server_socket = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
        if(server_socket == -1)return -1;
        setsockopt(server_socket, SOL_SOCKET, SO_REUSEADDR, (int[]){1}, sizeof(int));

        if(l4rb_bind_ipv4(server_socket, *(struct in_addr *)addr, port) == -1)
            goto fallback;

    }else if(inet_pton(AF_INET6, conn_str, addr) > 0){
        /*  IPv6    */
        server_socket = socket(AF_INET6, SOCK_STREAM | SOCK_NONBLOCK, 0);
        if(server_socket == -1)return -1;
        setsockopt(server_socket, SOL_SOCKET, SO_REUSEADDR, (int[]){1}, sizeof(int));

        if(l4rb_bind_ipv6(server_socket, *(struct in6_addr *)addr, port) == -1)
            goto fallback;
    }else return -1;

    if(listen(server_socket, QUEUE_LEN) == -1)
        goto fallback;
    l4rb->listen = server_socket;

    struct epoll_event ev = {
        .events = EPOLLIN,
        .data.u32 = L4RB_LISTEN_EV};
    if(epoll_ctl(l4rb->queue, EPOLL_CTL_ADD, server_socket, &ev) == -1)
        goto fallback;

    l4rb->inten_conn = L4RB_LISTEN_EV;
    l4rb->new_conn = 0;
    return server_socket;
fallback:
    close(server_socket);
    return -1;
}
uint32_t
l4rb_poll(struct L4rb *l4rb, int timeout){
    if(l4rb->inten_conn != INACTIVE_CONN)
        return l4rb->inten_conn;
    if(l4rb->queue_len == 0){
#ifdef HAVE_EPOLL
        l4rb->queue_len = epoll_wait(l4rb->queue,
                l4rb->ev, QUEUE_LEN, timeout);
#endif /* HAVE_EPOLL */
    }
    /*  Note: Level-Trigger */
    uint32_t curr_conn = INACTIVE_CONN;
    ssize_t wrote_size;
    uint32_t conn_segment;
    struct Ring *rb = l4rb->rb;
    struct L4conn *curr_l4conn;
    while(l4rb->queue_len > 0){
        l4rb->queue_len--;
        curr_conn = l4rb->ev[l4rb->queue_len].data.u32;
        /*  Accept  */
        if(curr_conn == L4RB_LISTEN_EV){
            socklen_t addr_len = sizeof(union Addr_in);
            curr_conn = l4rb->new_conn & l4rb->max_conn;
            int client = accept(l4rb->listen,
                    (struct sockaddr *)
                    &(l4rb->conn[curr_conn].addr),
                    &addr_len);
            if(client == -1)continue;

            l4rb->conn[curr_conn] =
                (struct L4conn){
                    .fd = client,
                    .rb_head = NO_SUCCEEDING,
                    .rb_tail = NO_SUCCEEDING};
            curr_conn = _l4rb_enqueue(l4rb,
                    l4rb->conn + curr_conn);
#ifndef recv_immediately_after_accept
            curr_conn = INACTIVE_CONN;
#endif
            if(curr_conn == INACTIVE_CONN)
                continue;
        }
        curr_l4conn = l4rb->conn + curr_conn;
        /*  Rx  */
        wrote_size = recv(
                curr_l4conn->fd,
                RING_TAIL_BUFFER(rb),
                RING_TAIL_SIZE(rb),
                MSG_DONTWAIT);

        if(wrote_size <= 0 || (conn_segment
                    = ring_rearrange(rb, wrote_size))
                == rb->tail){
            _l4rb_dequeue(l4rb, l4rb->conn + curr_conn);
            curr_conn = INACTIVE_CONN;
            continue;
        }
        /*  Ringbuffer droped a connection. */
        if(RING_SEG_CONN(rb, rb->tail) != INACTIVE_CONN){
            _l4rb_dequeue(l4rb, l4rb->conn + RING_SEG_CONN(rb, rb->tail));
            RING_SEG_CONN(rb, rb->tail) = INACTIVE_CONN;
        }
        /*  First receiving */
        if(curr_l4conn->rb_head == NO_SUCCEEDING){
            curr_l4conn->rb_head = conn_segment;
            curr_l4conn->rb_tail = conn_segment;
            RING_SEG_CONN(rb, conn_segment) = curr_conn;
        }
        ring_link_conn(rb, curr_l4conn->rb_tail, conn_segment);
        curr_l4conn->rb_tail = conn_segment;
    }
    return(l4rb->inten_conn = curr_conn);
}
